package demo.complex;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

// 订单类
class Order {
    /**
     * private static final：
     * private：限制该变量只能在类内部访问。
     * static：该变量属于类，而不是类的实例（所有对象共享同一个COUNTER）。
     * final：该变量初始化后不可被修改（引用不可变）。
     * AtomicInteger：
     * 是 Java java.util.concurrent.atomic 包中的类，提供原子操作（如自增、自减）。
     * 底层使用 CAS（Compare-and-Swap）机制，避免了传统锁（如synchronized）的性能开销。
     * new AtomicInteger(1000)：
     * 初始化计数器，初始值为 1000。后续生成的订单 ID 将从 1001 开始。
     */
    private static final AtomicInteger COUNTER = new AtomicInteger(1000);
    private final int orderId;
    private final String product;
    private final int quantity;
    private final long createTime;

    public Order(String product, int quantity) {
        /**
         * 原子性地将当前值加 1，并返回递增后的值。
         * 例如，若当前值为 1000，调用后返回 1001，同时内部值更新为 1001。
         * 线程安全：即使多个线程同时调用该方法，也不会出现竞态条件（如重复 ID）。
         */
        this.orderId = COUNTER.incrementAndGet();
        this.product = product;
        this.quantity = quantity;
        this.createTime = System.currentTimeMillis();
    }

    public int getOrderId() {
        return orderId;
    }

    public String getProduct() {
        return product;
    }

    public int getQuantity() {
        return quantity;
    }

    public long getCreateTime() {
        return createTime;
    }

    @Override
    public String toString() {
        return "Order{orderId=" + orderId +
                ", product='" + product + '\'' +
                ", quantity=" + quantity +
                ", createTime=" + createTime + '}';
    }
}

/**
 * 任务队列模式
 *      所有的生产者和消费者都共享一个系统，通过这个系统实现数据的共享
 *在原代码中：
 * 订单生成器将订单放入共享队列 orderQueue
 * 订单处理器从 orderQueue 中随机获取订单处理
 * 没有保证特定生产者生成的订单由特定消费者处理
 * 这种设计的优点是：
 * 高吞吐量：多个消费者可以并行处理订单
 * 负载均衡：任务自动分配给空闲的消费者
 * 系统简单：不需要维护生产者和消费者的映射关系
 * 但缺点是：
 * 无法保证特定订单由特定消费者处理
 * 难以实现订单处理的顺序性要求
 */
// 订单处理系统
class OrderProcessingSystem {
    private final BlockingQueue<Order> orderQueue = new LinkedBlockingQueue<>(100);
    private final BlockingQueue<Order> processedQueue = new LinkedBlockingQueue<>(100);

    // 提交订单
    public void submitOrder(Order order) throws InterruptedException {
        orderQueue.put(order);
        System.out.println("订单已提交: " + order.getOrderId());
    }

    // 处理订单
    public Order processOrder() throws InterruptedException {
        //当队列中没有数据时，take() 方法不会报错，而是会阻塞当前线程，直到队列中有元素可供取出。
        Order order = orderQueue.take();
        // 模拟订单处理逻辑
        Thread.sleep((long) (Math.random() * 1000));
        System.out.println("订单已处理: " + order.getOrderId() +
                ", 耗时: " + (System.currentTimeMillis() - order.getCreateTime()) + "ms");
        processedQueue.put(order);
        return order;
    }

    // 完成订单
    public Order completeOrder() throws InterruptedException {
        Order order = processedQueue.take();
        // 模拟订单完成逻辑
        System.out.println("订单已完成: " + order.getOrderId());
        return order;
    }
}

/**
 * 工作流模式
 * 方案选择建议
 * 如果需要严格隔离生产者和消费者：
 * 使用方案 1，为每个生产者分配独立的队列
 * 适用于需要保证特定订单流由特定处理链处理的场景
 * 例如：不同类型的订单需要不同的处理逻辑
 * 如果需要灵活的路由规则：
 * 使用方案 2，通过注册回调函数实现订单路由
 * 适用于需要动态分配处理逻辑的场景
 * 例如：根据订单金额、类型等属性决定由哪个消费者处理
 * 如果原有的任务队列模式仍能满足需求：
 * 保持原设计不变
 * 适用于不需要关心哪个消费者处理哪个订单的场景
 * 例如：批处理任务、后台计算任务等
 */
// 订单处理系统（方案1）
class OrderProcessingSystemV1 {
    //生产者数量
    private final int producerCount;
    private final BlockingQueue<Order>[] orderQueues;
    private final BlockingQueue<Order> processedQueue = new LinkedBlockingQueue<>(100);

    @SuppressWarnings("unchecked")
    public OrderProcessingSystemV1(int producerCount) {
        this.producerCount = producerCount;
        this.orderQueues = new BlockingQueue[producerCount];
        for (int i = 0; i < producerCount; i++) {
            orderQueues[i] = new LinkedBlockingQueue<>(10);
        }
    }

    // 提交订单到指定队列
    public void submitOrder(Order order, int producerId) throws InterruptedException {
        orderQueues[producerId % producerCount].put(order);
        System.out.println("生产者" + producerId + "提交订单: " + order.getOrderId());
    }

    // 从指定队列处理订单
    public Order processOrder(int consumerId) throws InterruptedException {
        Order order = orderQueues[consumerId % producerCount].take();
        // 处理逻辑...
        System.out.println("消费者" + consumerId + "处理订单: " + order.getOrderId());
        processedQueue.put(order);
        return order;
    }

    // 完成订单（保持不变）
    public Order completeOrder() throws InterruptedException {
        Order order = processedQueue.take();
        System.out.println("订单已完成: " + order.getOrderId());
        return order;
    }
}

// 订单处理系统（方案2）
class OrderProcessingSystemV2 {
    private final BlockingQueue<Order> orderQueue = new LinkedBlockingQueue<>(100);
    private final BlockingQueue<Order> processedQueue = new LinkedBlockingQueue<>(100);
    //Hashtable	全表锁（synchronized）	同一时间仅 1 个线程可操作
    //Collections.synchronizedMap	全表锁（synchronized）	同一时间仅 1 个线程可操作
    //ConcurrentHashMap（JDK 7）	分段锁（Segment）	默认 16 个线程可并行
    //ConcurrentHashMap（JDK 8+）	锁节点（synchronized）	理论上支持无限线程并行（只要不冲突）
    //ConcurrentHashMap 在多线程写入时，会针对同一个哈希槽的头节点加锁，而同一个 key 必然落在同一个哈希槽中，因此对同一 key 的写入操作会互斥

    //场景	HashMap（线程不安全）	ConcurrentHashMap（线程安全）
    //多线程修改同一 key	可能在覆盖过程中因指针竞争导致链表结构损坏，或中间状态被其他线程读取。	通过锁机制确保覆盖操作原子性，其他线程读取时只能看到完整修改后的状态，不会读取到中间值。
    //复合操作安全性	put操作不是原子的，可能被其他线程中断（如插入时被删除）。	put操作整体是原子的，要么完全成功，要么失败（不影响数据结构）。
    //扩容时的覆盖	扩容过程中链表迁移可能被打断，导致循环链表或节点丢失。	扩容分段进行，且迁移时通过标记节点隔离并发操作，覆盖操作不会干扰扩容流程。
    private final ConcurrentHashMap<Integer, Consumer<Order>> consumerMap = new ConcurrentHashMap<>();

    // 注册消费者处理特定生产者的订单
    public void registerConsumer(int producerId, Consumer<Order> consumer) {
        consumerMap.put(producerId, consumer);
    }

    // 提交订单（根据生产者ID路由）
    public void submitOrder(Order order, int producerId) throws InterruptedException {
        System.out.println("生产者" + producerId + "提交订单: " + order.getOrderId());

        // 方式1：直接路由给指定消费者
        Consumer<Order> consumer = consumerMap.get(producerId);
        if (consumer != null) {
            consumer.accept(order);
            return;
        }

        // 方式2：放入公共队列（当没有指定消费者时）
        orderQueue.put(order);
    }

    // 从公共队列处理订单
    public Order processOrder() throws InterruptedException {
        Order order = orderQueue.take();
        // 处理逻辑...
        System.out.println("处理公共订单: " + order.getOrderId());
        processedQueue.put(order);
        return order;
    }

    // 完成订单（保持不变）
    public Order completeOrder() throws InterruptedException {
        Order order = processedQueue.take();
        System.out.println("订单已完成: " + order.getOrderId());
        return order;
    }
}

// 订单生成器（生产者）
class OrderGenerator implements Runnable {
    private final OrderProcessingSystem system;
    private final int id;
    private final String[] products = {"手机", "电脑", "平板", "耳机", "相机"};

    public OrderGenerator(OrderProcessingSystem system, int id) {
        this.system = system;
        this.id = id;
    }

    @Override
    public void run() {
        try {
            for (int i = 0; i < 5; i++) {
                String product = products[(int) (Math.random() * products.length)];
                int quantity = (int) (Math.random() * 10) + 1;
                Order order = new Order(product, quantity);
                system.submitOrder(order);
                Thread.sleep((long) (Math.random() * 500));
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

// 订单处理器（消费者1）
class OrderProcessor implements Runnable {
    private final OrderProcessingSystem system;
    private final int id;

    public OrderProcessor(OrderProcessingSystem system, int id) {
        this.system = system;
        this.id = id;
    }

    @Override
    public void run() {
        try {
            for (int i = 0; i < 5; i++) {
                system.processOrder();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

// 订单完成器（消费者2）
class OrderFinalizer implements Runnable {
    private final OrderProcessingSystem system;
    private final int id;

    public OrderFinalizer(OrderProcessingSystem system, int id) {
        this.system = system;
        this.id = id;
    }

    @Override
    public void run() {
        try {
            for (int i = 0; i < 5; i++) {
                system.completeOrder();
                Thread.sleep((long) (Math.random() * 300));
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

public class ComplexBusinessDemo {
    public static void main(String[] args) {
        OrderProcessingSystem system = new OrderProcessingSystem();

        // 创建生产者和消费者线程
        Thread generator1 = new Thread(new OrderGenerator(system, 1));
        Thread generator2 = new Thread(new OrderGenerator(system, 2));
        Thread processor1 = new Thread(new OrderProcessor(system, 1));
        Thread processor2 = new Thread(new OrderProcessor(system, 2));
        Thread finalizer1 = new Thread(new OrderFinalizer(system, 1));
        Thread finalizer2 = new Thread(new OrderFinalizer(system, 2));

        // 启动线程
        generator1.start();
        generator2.start();
        processor1.start();
        processor2.start();
        finalizer1.start();
        finalizer2.start();

        // 等待所有线程完成
        try {
            generator1.join();
            generator2.join();
            processor1.join();
            processor2.join();
            finalizer1.join();
            finalizer2.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("订单处理系统演示完成");
    }
}